/*
Copyright 2024 egonlin.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package controller

import (
	"context"
	"fmt"
	"net"
	"os"
	"time"

	"github.com/go-logr/logr"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	databasev1 "mysql-operator/api/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

// 会根据下述标识生成RBAC
// +kubebuilder:rbac:groups="",resources=pods;pods/exec,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=db.example.com,resources=mysqlclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=db.example.com,resources=mysqlclusters/status,verbs=get;update;patch

// Define global constants for MySQL password and kubeconfig path
const (
	MySQLPassword          = "password"           // Hardcoded MySQL password
	KubeConfigPath         = "/root/.kube/config" // Hardcoded kubeconfig path
	MysqlClusterKind       = "MysqlCluster"
	MysqlClusterAPIVersion = "apps.egonlin.com/v1"
)

type MysqlClusterReconciler struct {
	client.Client
	Log    logr.Logger
	Scheme *runtime.Scheme
}

func (r *MysqlClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx)
	log.Info("调谐函数触发执行", "req", req) // 额外增加1个字段

	var cluster databasev1.MysqlCluster
	if err := r.Get(ctx, req.NamespacedName, &cluster); err != nil {
		if errors.IsNotFound(err) {
			log.Info("MysqlCluster resource not found. Ignoring since object must be deleted")
			return ctrl.Result{}, nil
		}
		log.Error(err, "Failed to get MysqlCluster")
		return ctrl.Result{}, err
	}

	// Get or create MySQL master pod
	masterPod := &v1.Pod{}
	masterPodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Master}
	if err := r.Get(ctx, masterPodKey, masterPod); err != nil {
		if errors.IsNotFound(err) {
			masterPod = r.createPod(cluster.Spec.Master, "master", cluster.Spec.MasterImage, cluster.Spec.MasterConfigMap, cluster.Namespace, cluster)
			if err := r.Create(ctx, masterPod); err != nil {
				log.Error(err, "Failed to create MySQL master pod")
				return ctrl.Result{}, err
			}
		} else {
			log.Error(err, "Failed to get MySQL master pod")
			return ctrl.Result{}, err
		}
	}

	// Get or create MySQL slave pod
	slavePod := &v1.Pod{}
	slavePodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Slave}
	if err := r.Get(ctx, slavePodKey, slavePod); err != nil {
		if errors.IsNotFound(err) {
			slavePod = r.createPod(cluster.Spec.Slave, "slave", cluster.Spec.SlaveImage, cluster.Spec.SlaveConfigMap, cluster.Namespace, cluster)
			if err := r.Create(ctx, slavePod); err != nil {
				log.Error(err, "Failed to create MySQL slave pod")
				return ctrl.Result{}, err
			}
		} else {
			log.Error(err, "Failed to get MySQL slave pod")
			return ctrl.Result{}, err
		}
	}

	// Update Service to point to current Master Pod
	if err := r.updateService(ctx, cluster); err != nil {
		log.Error(err, "Failed to update MySQL service")
		return ctrl.Result{}, err
	}

	// Initialize replication if not already done
	if cluster.Status.Master == "" || cluster.Status.Slave == "" {
		log.Info("Initializing MySQL replication")
		if err := r.initializeReplication(ctx, cluster); err != nil {
			log.Error(err, "Failed to initialize MySQL replication")
			return ctrl.Result{}, err
		}
		cluster.Status.Master = cluster.Spec.Master
		cluster.Status.Slave = cluster.Spec.Slave
		cluster.Status.Ready = true
		if err := r.Status().Update(ctx, &cluster); err != nil {
			log.Error(err, "Failed to update MysqlCluster status")
			return ctrl.Result{}, err
		}
	}

	// 通过检测service是否可以正常访问来判定主库是否挂掉
	masterAlive, err := r.isServiceAlive(ctx, cluster)
	if err != nil {
		log.Error(err, "Failed to check master service status")
		return ctrl.Result{}, err
	} else {
		log.Info("当前主库存活状态", "masterAlive", masterAlive)

	}

	if !masterAlive {
		log.Info("Master service is down, promoting slave to master")
		if err := r.handleFailover(ctx, cluster); err != nil {
			log.Error(err, "Failed to handle failover")
			return ctrl.Result{}, err
		}
		cluster.Status.Failover = true
		if err := r.Status().Update(ctx, &cluster); err != nil {
			log.Error(err, "Failed to update MysqlCluster status")
			return ctrl.Result{}, err
		}
	} else {
		cluster.Status.Failover = false
		if err := r.Status().Update(ctx, &cluster); err != nil {
			log.Error(err, "Failed to update MysqlCluster status")
			return ctrl.Result{}, err
		}
	}
	return ctrl.Result{RequeueAfter: time.Minute}, nil
}

func (r *MysqlClusterReconciler) createPod(name, role, image string, configMapName, namespace string, cluster databasev1.MysqlCluster) *v1.Pod {
	//定义 OwnerReference，你可以去创建一个deployment然后查看它的pod的yaml中就有metadata.ownerReferences，这个非常字段非常关键
	// 有了该字段，并且明确配置了如下APIVersion、Kind、Name、UID、Controller后，才会与控制器关联上，文末的SetupWithManager 函数中Owns(&v1.Pod{})才会有意义：pod的相关事件才会
	// 通知给调谐函数
	ownerRef := metav1.OwnerReference{
		APIVersion: MysqlClusterAPIVersion, // 使用常量
		Kind:       MysqlClusterKind,       // 使用常量
		Name:       cluster.Name,
		UID:        cluster.UID,
		Controller: func(b bool) *bool { return &b }(true), // 使用匿名函数创建布尔值指针, // 指示此 OwnerReference 是控制器，这一条非常关键
	}
	return &v1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
			Labels: map[string]string{
				"app":  "mysql",
				"role": role,
			},
			OwnerReferences: []metav1.OwnerReference{
				ownerRef,
			},
		},
		Spec: v1.PodSpec{
			Containers: []v1.Container{
				{
					Name:  "mysql",
					Image: image,
					Env: []v1.EnvVar{
						{
							Name:  "MYSQL_ROOT_PASSWORD",
							Value: "password",
						},
					},
					Ports: []v1.ContainerPort{
						{
							Name:          "mysql",
							ContainerPort: 3306,
						},
					},
					VolumeMounts: []v1.VolumeMount{
						{
							Name:      "mysql-config",
							MountPath: "/etc/my.cnf",
							SubPath:   "my.cnf", // 使用 SubPath 来挂载文件
						},
					},
				},
			},
			Volumes: []v1.Volume{
				{
					Name: "mysql-config",
					VolumeSource: v1.VolumeSource{
						ConfigMap: &v1.ConfigMapVolumeSource{
							LocalObjectReference: v1.LocalObjectReference{
								Name: configMapName,
							},
						},
					},
				},
			},
		},
	}
}

func (r *MysqlClusterReconciler) initializeReplication(ctx context.Context, cluster databasev1.MysqlCluster) error {
	masterPod := &v1.Pod{}
	masterPodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Master}
	if err := r.Get(ctx, masterPodKey, masterPod); err != nil {
		return err
	}

	slavePod := &v1.Pod{}
	slavePodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Slave}
	if err := r.Get(ctx, slavePodKey, slavePod); err != nil {
		return err
	}

	// Configure replication using Kubernetes exec
	// err := r.execCommandOnPod(masterPod, "mysql -uroot -p'%s' -e 'CREATE USER IF NOT EXISTS `replica`@`%` IDENTIFIED BY `password`; GRANT REPLICATION SLAVE ON *.* TO `replica`@`%`;'")
	masterCommand := fmt.Sprintf(
		"mysql -uroot -p%s -e \"CREATE USER IF NOT EXISTS 'replica'@'%%' IDENTIFIED BY 'password'; GRANT REPLICATION SLAVE ON *.* TO 'replica'@'%%';\"",
		MySQLPassword,
	)
	err := r.execCommandOnPod(masterPod, masterCommand)
	if err != nil {
		return err
	}

	masterServiceName := cluster.Spec.ServiceName
	slaveCommand := fmt.Sprintf(
		"mysql -uroot -p%s -e \"CHANGE MASTER TO MASTER_HOST='%s', MASTER_USER='replica', MASTER_PASSWORD='password', MASTER_AUTO_POSITION=1;; START SLAVE;\"",
		MySQLPassword,
		masterServiceName,
	)
	err = r.execCommandOnPod(slavePod, slaveCommand)
	return err
}

func (r *MysqlClusterReconciler) execCommandOnPod(pod *v1.Pod, command string) error {
	// Load kubeconfig from default location
	//kubeconfig := os.Getenv("KUBECONFIG")
	//if kubeconfig == "" {
	//	kubeconfig = "/root/.kube/config" // Fallback to default path
	//}
	config, err := clientcmd.BuildConfigFromFlags("", KubeConfigPath)
	if err != nil {
		return err
	}

	// Create a new Kubernetes clientset
	kubeClient, err := kubernetes.NewForConfig(config)
	if err != nil {
		return err
	}

	// Create REST client for pod exec
	restClient := kubeClient.CoreV1().RESTClient()
	req := restClient.
		Post().
		Resource("pods").
		Name(pod.Name).
		Namespace(pod.Namespace).
		SubResource("exec").
		Param("stdin", "false").
		Param("stdout", "true").
		Param("stderr", "true").
		Param("tty", "false").
		Param("container", pod.Spec.Containers[0].Name).
		Param("command", "/bin/sh").
		Param("command", "-c").
		Param("command", command)

	// Create an executor
	executor, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
	if err != nil {
		return err
	}

	// Execute the command
	return executor.Stream(remotecommand.StreamOptions{
		Stdout: os.Stdout,
		Stderr: os.Stderr,
	})
}

func (r *MysqlClusterReconciler) updateService(ctx context.Context, cluster databasev1.MysqlCluster) error {
	// Get or create the Service
	service := &v1.Service{}
	serviceKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.ServiceName}

	if err := r.Get(ctx, serviceKey, service); err != nil {
		if errors.IsNotFound(err) {
			// Create service object if it does not exist
			service = r.createService(cluster)

			// 调用r.Create把上面生存的service对象提交给k8s
			if err := r.Create(ctx, service); err != nil {
				return err
			}
		} else {
			return err
		}
	}

	return nil
}

func (r *MysqlClusterReconciler) createService(cluster databasev1.MysqlCluster) *v1.Service {
	return &v1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      cluster.Spec.ServiceName,
			Namespace: cluster.Namespace,
			Labels: map[string]string{
				"app": "mysql",
			},
		},
		Spec: v1.ServiceSpec{
			Selector: map[string]string{
				"role": "master", // 默认选择主库
			},
			Ports: []v1.ServicePort{
				{
					Name: "mysql",
					Port: 3306,
				},
			},
		},
	}
}

func (r *MysqlClusterReconciler) handleFailover(ctx context.Context, cluster databasev1.MysqlCluster) error {
	// Promote slave to master
	slavePod := &v1.Pod{}
	slavePodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Slave}
	if err := r.Get(ctx, slavePodKey, slavePod); err != nil {
		return err
	}

	masterPod := &v1.Pod{}
	masterPodKey := client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Spec.Master}
	if err := r.Get(ctx, masterPodKey, masterPod); err != nil {
		return err
	}

	// Execute commands to promote the slave to master and reconfigure the old master
	slaveCommand := fmt.Sprintf(
		"mysql -uroot -p%s -e 'STOP SLAVE; SET GLOBAL read_only = OFF;'",
		MySQLPassword,
	)

	if err := r.execCommandOnPod(slavePod, slaveCommand); err != nil {
		return err
	}

	// Update status
	cluster.Status.Master = cluster.Spec.Slave
	cluster.Status.Slave = cluster.Spec.Master
	cluster.Status.Ready = true

	// Update service to point to the new master
	if err := r.updateService(ctx, cluster); err != nil {
		return err
	}

	// Reconfigure old master as slave
	oldMasterPod := &v1.Pod{}
	if err := r.Get(ctx, masterPodKey, oldMasterPod); err != nil {
		return err
	}

	if err := r.execCommandOnPod(oldMasterPod, fmt.Sprintf("mysql -e 'CHANGE MASTER TO MASTER_HOST=\"%s\", MASTER_USER=\"replica\", MASTER_PASSWORD=\"password\", MASTER_LOG_FILE=\"mysql-bin.000001\", MASTER_LOG_POS=4:; START SLAVE;'", cluster.Spec.Slave)); err != nil {
		return err
	}

	return nil
}

func (r *MysqlClusterReconciler) isPodAlive(ctx context.Context, podName string) (bool, error) {
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:3306", podName), 5*time.Second)
	if err != nil {
		return false, nil // Port not reachable
	}
	conn.Close()
	return true, nil
}

func (r *MysqlClusterReconciler) isServiceAlive(ctx context.Context, cluster databasev1.MysqlCluster) (bool, error) {
	// 通过检测service的endpoints字段是否为空来判定service关联的主库是否挂掉了，这是最快的检测方式
	svcName := cluster.Spec.ServiceName
	namespace := cluster.Namespace

	svc := &v1.Service{}
	if err := r.Get(ctx, types.NamespacedName{Name: svcName, Namespace: namespace}, svc); err != nil {
		return false, err
	}

	// 获取 Service 对应的 Endpoints
	endpoints := &v1.Endpoints{}
	if err := r.Get(ctx, types.NamespacedName{Name: svcName, Namespace: namespace}, endpoints); err != nil {
		return false, err
	}

	// 检查 Endpoints 中是否有地址，如果没有，则表示 Service 对应的 Pod 不可用
	for _, subset := range endpoints.Subsets {
		if len(subset.Addresses) > 0 {
			return true, nil
		}
	}

	return false, nil
}

func (r *MysqlClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	// 增加：Owns(&v1.Pod{})，确保检测pod资源
	return ctrl.NewControllerManagedBy(mgr).
		For(&databasev1.MysqlCluster{}).
		Owns(&v1.Pod{}).
		Complete(r)
}

